/*
 * BSM6.0
 *
 * Copyright 2012-2013 (C) SINODATA CO., LTD. All Rights Reserved.
 */
package com.sinodata.bsm.inmp.service;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;

import org.apache.log4j.Logger;

import com.sinodata.bsm.common.utils.ThreadPool;
import com.sinodata.bsm.inmp.conf.EventDispatcher;
import com.sinodata.bsm.inmp.util.SocketUtil;
import com.sinodata.bsm.inmp.util.SystemContext;

/**
 * <p>
 * Description: 启动一个socket端口    
 * </p>
 *
 * @author wangyazhou
 * @version 1.0

 * <p>
 * History: 
 *
 * Date                     Author         Version     Description
 * ---------------------------------------------------------------------------------
 * 2012-8-10 下午9:32:39          wangyazhou        1.0         To create
 * </p>
 *
 * @since   
 * @see     
 */
public class SocketService extends Thread {

    private static final Logger logger = Logger.getLogger(SocketService.class);

    static final String EVENT = "EVENT";

    public SocketService() {
        logger.info("[TIP]\t init");
    }

    @Override
    @SuppressWarnings("rawtypes")
    public void run() {
        Selector selector;
        ServerSocketChannel ssc;
        int port = Integer.parseInt(SystemContext.getProperty("socket").toString()); //tdd
        try {
            selector = Selector.open();
            ssc = ServerSocketChannel.open();
            InetSocketAddress address = new InetSocketAddress(port);
            ssc.socket().bind(address);
            ssc.configureBlocking(false);
            ssc.register(selector, SelectionKey.OP_ACCEPT);
            logger.info("Probe listener service start,port" + port);
        } catch (IOException e1) {
            logger.error("failed to start listener service", e1);
            return;
        }

        Socket socket = null;
        while (true) {
            try {
                if (selector.select() == 0) {
                    continue;
                }
                Iterator i = selector.selectedKeys().iterator();
                while (i.hasNext()) {
                    SelectionKey key = (SelectionKey) i.next();
                    i.remove();
                    if (key.isAcceptable()) {
                        socket = ((ServerSocketChannel) key.channel()).accept().socket();
                        ThreadPool.getInstance().execute(new RequestProcessor(socket));
                    }
                }
            } catch (IOException ex) {
                logger.error("failed to send socket info", ex);
                SocketUtil.closeSocket(socket);
            }
        }
    }

    /**
     * 把json字符串转换为事件对象   并返回 1:成功  0：失败
     */
    private class RequestProcessor implements Runnable {
        private final Socket socket;

        public RequestProcessor(Socket socket) {
            this.socket = socket;
        }

        public void run() {
            Object object = null;
            try {
                object = SocketUtil.readObject(socket);
            } catch (Exception e) {
                SocketUtil.closeSocket(socket);
                logger.error("Illegal request", e);
                return;
            }
            if (object == null) {
                return;
            }

            EventDispatcher dispatcher = new EventDispatcher(object);
            dispatcher.run();
        }
    }

}
